# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import json
import time
from typing import List
from pydantic import TypeAdapter
from exception import AppException
from common.utils.urls import UrlUtil
from common.enums.models import ModelsEnum
from common.models.dev import DevModelsModel
from apps.admin.schemas.setting.ai import models_schema as schema


class AiModelsService:
    """ 模型配置服务类 """

    @classmethod
    async def channel(cls) -> schema.AiModelsChannelVo:
        """
        模型通道

        Returns:
            schema.AiModelsChannelVo

        Author:
            zero
        """
        chat_models = ModelsEnum.chat_models()
        vector_models = ModelsEnum.vector_models()
        ranking_models = ModelsEnum.ranking_models()

        for item in chat_models:
            chat_models[item]["logo"] = await UrlUtil.to_absolute_url(chat_models[item]["logo"])

        for item in vector_models:
            vector_models[item]["logo"] = await UrlUtil.to_absolute_url(vector_models[item]["logo"])

        return schema.AiModelsChannelVo(
            chat_models=chat_models,
            vector_models=vector_models,
            ranking_models=ranking_models
        )

    @classmethod
    async def lists(cls, scene: str) -> List[schema.AiModelsListVo]:
        """
        模型列表

        Args:
            scene (str): 场景

        Returns:
            List[schema.AiModelsListVo]

        Author:
            zero
        """
        lists = await (DevModelsModel
                       .filter(scene=scene, is_delete=0)
                       .order_by("sort", "is_disable", "-id")
                       .all()
                       .values(*("id,logo,channel,name,sort,is_disable,config,models,is_builtin".split(","))))

        _lists = []
        for item in lists:
            models = json.loads(item['models'] or "[]")
            configs = json.loads(item['config'] or "{}")
            channel = ModelsEnum.get_model_conf(scene, item["channel"]).get("name", "")

            item["channel"] = channel
            item["models_num"] = len(models)
            item["agency_api"] = 1 if configs.get("agency_api") else 0
            item["logo"] = await UrlUtil.to_absolute_url(item["logo"])
            _lists.append(TypeAdapter(schema.AiModelsListVo).validate_python(item))

        return _lists

    @classmethod
    async def detail(cls, id_: int) -> schema.AiModelsDetailVo:
        """
        模型详情

        Args:
            id_ (int): 模型ID

        Returns:
            schema.AiModelsDetailVo

        Author:
            zero
        """
        detail = await DevModelsModel.filter(id=id_, is_delete=0).get()
        config = json.loads(detail.config)
        models = json.loads(detail.models)

        setting = ModelsEnum.get_model_conf(detail.scene, detail.channel)
        for conf in setting["configs"]:
            if conf.get("config"):
                for item in conf["config"]:
                    key = item["key"]
                    val = item["default"] if config.get(key) is None else config.get(key)
                    item["default"] = val
            else:
                key = conf["key"]
                val = conf["default"] if config.get(key) is None else config.get(key)
                val = bool(val) if conf["type"] == "switch" else val
                conf["default"] = val

        return schema.AiModelsDetailVo(
            id=detail.id,
            scene=detail.scene,
            mark=detail.mark,
            channel=detail.channel,
            logo=await UrlUtil.to_absolute_url(detail.logo),
            name=detail.name,
            intro=detail.intro,
            is_disable=detail.is_disable,
            config=setting["configs"],
            models=models
        )

    @classmethod
    async def add(cls, post: schema.AiModelsAddIn) -> None:
        """
        模型新增

        Args:
            post (schema.AiModelsAddIn): 参数

        Author:
            zero
        """
        configs, models = cls._handle_llm(post.scene, post.channel, post.config, post.models)
        await DevModelsModel.create(
            scene=post.scene,
            channel=post.channel,
            mark=post.mark,
            logo=UrlUtil.to_relative_url(post.logo),
            name=post.name,
            intro=post.intro,
            config=json.dumps(configs, ensure_ascii=False),
            models=json.dumps(models, ensure_ascii=False),
            is_disable=post.is_disable,
            create_time=int(time.time()),
            update_time=int(time.time())
        )

    @classmethod
    async def edit(cls, post: schema.AiModelsEditIn) -> None:
        """
        模型编辑

        Args:
            post (schema.AiModelsEditIn): 参数

        Author:
            zero
        """
        detail = await DevModelsModel.filter(id=post.id, is_delete=0).get()
        configs, models = cls._handle_llm(detail.scene, post.channel, post.config, post.models)
        await DevModelsModel.filter(id=post.id).update(
            channel=post.channel,
            logo=UrlUtil.to_relative_url(post.logo),
            name=post.name,
            intro=post.intro,
            config=json.dumps(configs, ensure_ascii=False),
            models=json.dumps(models, ensure_ascii=False),
            is_disable=post.is_disable,
            update_time=int(time.time())
        )

    @classmethod
    async def delete(cls, id_: int):
        """
        模型删除

        Args:
            id_ (int): 模型ID

        Author:
            zero
        """
        detail = await DevModelsModel.filter(id=id_, is_delete=0).get()
        await (DevModelsModel
               .filter(id=detail.id)
               .update(is_delete=1, delete_time=int(time.time())))

    @classmethod
    async def disable(cls, id_: int) -> None:
        """
        模型禁用

        Args:
            id_ (int): 模型ID

        Author:
            zero
        """
        detail = await DevModelsModel.filter(id=id_, is_delete=0).get()
        await (DevModelsModel
               .filter(id=detail.id)
               .update(
                    is_disable=0 if detail.is_disable else 1,
                    update_time=int(time.time()))
               )

    @classmethod
    def _handle_llm(cls, scene: str, channel: str, config: dict, models: list) -> tuple:
        """
        模型处理 (验证模型数据)

        Args:
            scene (str): 场景
            channel (str): 通道
            config (dict): 通道配置参数
            models (list): 子级模型列表

        Returns:
            tuple

        Author:
            zero
        """
        setting = ModelsEnum.get_model_conf(scene, channel)
        if not setting:
            raise AppException("AI模型通道异常")

        _configs = {}
        for conf in setting["configs"]:
            if conf.get("config"):
                for item in conf.get("config"):
                    key = item["key"]
                    val = item["default"] if config.get(key) is None else config.get(key)
                    _configs[key] = val
                    if item.get("range") and item.get("type") == "slider":
                        if val < item["range"][0] or val > item["range"][1]:
                            raise AppException(f"{key}配置不在合法值内")
            else:
                key = conf["key"]
                val = conf["default"] if config.get(key) is None else config.get(key)
                _configs[key] = val

        _exist = []
        _models = []
        for llm in models:
            if not llm.get("model"):
                raise AppException("请设置模型名称")
            if not llm.get("alias"):
                raise AppException("请设置模型别名")
            if llm.get("modal", 0) not in [0, 1]:
                raise AppException("请设置模态状态")
            if llm.get("status", 0) not in [0, 1]:
                raise AppException("请设置模型状态")
            if llm.get("model", "") in _exist:
                raise AppException("模型不可重复添加: " + llm.get("model"))

            try:
                price = int(llm.get("price") or 0)
            except (TypeError, ValueError):
                raise AppException("请正确输入模型价格")

            _exist.append(llm.get("model").strip())
            if scene == ModelsEnum.TYPE_CHAT:
                _models.append({
                    "model": llm.get("model").strip(),
                    "alias": llm.get("alias").strip(),
                    "price": price,
                    "modal": int(llm.get("modal", "0")),
                    "status": int(llm.get("status", "0")),
                    "max_context": int(llm.get("max_context", "0")),
                    "max_output": int(llm.get("max_output", "0"))
                })
            else:
                _models.append({
                    "model": llm.get("model").strip(),
                    "alias": llm.get("alias").strip(),
                    "price": price,
                    "modal": int(llm.get("modal", "0")),
                    "status": int(llm.get("status", "0")),
                    "dimension": int(llm.get("dimension", "1536"))
                })

        return _configs, _models
